package com.promote.hotspot.server.launcher;

import com.promote.hotspot.server.netty.dashboard.NettyClient;
import com.promote.hotspot.server.rule.KeyRuleHolder;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Watch;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.promote.hotspot.common.config.ConfigConstant;
import org.promote.hotspot.common.etcd.JetcdClient;
import org.promote.hotspot.common.rule.KeyRule;
import org.promote.hotspot.common.tool.FastJsonUtils;
import org.promote.hotspot.common.tool.IpUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.*;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.promote.hotspot.common.etcd.JetcdClient.bytesOf;

/**
 * 服务端启动类
 *
 * @author enping.jep
 * @date 2023/10/19 15:05
 **/

@Slf4j
@Component
public class EtcdLauncher {
    @Resource
    private JetcdClient jetcdClient;

    /**
     * 该server放到etcd server目录的哪个app下
     */
    @Value("${etcd.serverPath}")
    private String serverPath;

    @Value("${local.address}")
    private String localAddress;

    @Value("${netty.port}")
    private int port;

    //默认目录
    private static final String DEFAULT_PATH = "default";

    /**
     * 监听规则变化
     */
    @PostConstruct
    private void watchRules() {
        log.info("into watchRules");
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                String key = ConfigConstant.RULE_PATH;
                log.info(getClass() + "---server begin watch rule change ----");
                jetcdClient.watchPrefix(key, Watch.listener(watchResponse -> {
                    log.info("收到[{}]的事件", key);
                    // 被调用时传入的是事件集合，这里遍历每个事件
                    watchResponse.getEvents().forEach(watchEvent -> {
                        // 操作类型
                        WatchEvent.EventType eventType = watchEvent.getEventType();
                        // 操作的键值对
                        KeyValue keyValue = watchEvent.getKeyValue();
                        log.info("type={}, key={}, value={}",
                                eventType,
                                keyValue.getKey().toString(UTF_8),
                                keyValue.getValue().toString(UTF_8));
                        ruleChange(keyValue);
                    });
                }));
            }
        });
    }

    /**
     * 每隔一分钟拉取一次rule
     */
    @PostConstruct
    private void pullRules() {
        log.info("init pullRules");
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        ScheduledFuture<?> future = scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                if (isForSingle()) {
                    String value = jetcdClient.get(ConfigConstant.RULE_PATH + serverPath);
                    if (StringUtils.isNotEmpty(value)) {
                        List<KeyRule> keyRules = FastJsonUtils.toList(value, KeyRule.class);
                        KeyRuleHolder.put(serverPath, keyRules);
                    }
                } else {
                    List<KeyValue> keyValues = jetcdClient.getPrefix(ConfigConstant.RULE_PATH);
                    for (KeyValue keyValue : keyValues) {
//                        log.info("key={}, value={}",
//                                keyValue.getKey().toString(UTF_8),
//                                keyValue.getValue().toString(UTF_8));
                        ruleChange(keyValue);
                    }
                }
            }
        }, 0, 60, TimeUnit.SECONDS);

        /**  先注释掉，这个代码可以用来捕获异常，但会影响初始化的过程
         try {
         future.get();
         } catch (InterruptedException | ExecutionException t) {
         t.printStackTrace();
         }
         */

    }

    @PostConstruct
    private void heartBeat() {
        log.info("init heartBeat");
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        //和etcd保持心跳连接
        scheduledExecutorService.scheduleAtFixedRate(this::uploadServerInfo, 0, 5, TimeUnit.SECONDS);
    }


    /**
     * 每隔5s，重新进行一次上报操作
     */
    public void uploadServerInfo() {
        jetcdClient.putWithLease(buildKey(), buildValue(), 10);//租约
    }

    /**
     * @return server端在etcd中的目录信息
     */
    private String buildKey() {
        String hostName = IpUtils.getHostName();
        return ConfigConstant.SERVERS_PATH + serverPath + "/" + hostName;
    }

    /**
     * @return 本机地址和端口
     */
    private String buildValue() {
        String ip;

        if (StringUtils.isNotEmpty(localAddress)) {
            ip = localAddress;
        } else {
            ip = IpUtils.getIp();
        }
        return ip + ":" + port;
    }

    /**
     * rule发生变化时，更新缓存的rule
     */
    private synchronized void ruleChange(KeyValue keyValue) {
        //原始路径为：/hotspot/rules/test1
        String appName = keyValue.getKey().toString(UTF_8).replace(ConfigConstant.RULE_PATH, "");
        if (StringUtils.isEmpty(appName)) {
            return;
        }
        String ruleJson = keyValue.getValue().toString(UTF_8);
        List<KeyRule> keyRules = FastJsonUtils.toList(ruleJson, KeyRule.class);
        KeyRuleHolder.put(appName, keyRules);
    }

    /**
     * 该server是否只服务于一个应用
     */
    private boolean isForSingle() {
        return !DEFAULT_PATH.equals(serverPath);
    }

    @PostConstruct
    public void fetchDashboardSchedule() {
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        //和etcd保持心跳连接
        scheduledExecutorService.scheduleAtFixedRate(this::fetchDashboard, 0, 30, TimeUnit.SECONDS);
    }

    public void fetchDashboard() {
        try {
            //获取DashboardIp
            List<KeyValue> keyValues = jetcdClient.getPrefix(ConfigConstant.DASH_BOARD_PATH);

            //是空，给个警告
            if (CollectionUtils.isEmpty(keyValues)) {
                log.warn("very important warn !!! Dashboard ip is null!!!");
                return;
            }

            String dashboardIp = keyValues.get(0).getValue().toString(UTF_8);
            NettyClient.getInstance().connect(dashboardIp);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
